package com.geotmt.billingcenter.config.kafka;

import com.alibaba.fastjson.JSON;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.kafka.core.KafkaTemplate;
import org.springframework.kafka.support.SendResult;
import org.springframework.stereotype.Component;
import org.springframework.util.concurrent.ListenableFuture;
import org.springframework.util.concurrent.ListenableFutureCallback;

import javax.annotation.Resource;

/**
 * @Description kafka 生产者
 * @Author      yanghanwei
 * @Mail        yanghanwei@geotmt.com
 * @Date        2019/3/16 17:33
 * @Version     v1
 **/
@Component
public class KafkaSender<T> {

    private Logger logger = LoggerFactory.getLogger(KafkaSender.class);

    @Value("${spring.kafka.open}")
    private boolean open;

    @Resource
    private KafkaTemplate<String, Object> kafkaTemplate;

    /**
     * kafka 发送消息
     * @param topic 主题
     * @param obj   消息主体
     */
    public void send(String topic,T obj) {

        String jsonObj = JSON.toJSONString(obj);

        logger.info("topic:{},msg:{}",topic, jsonObj);
        //发送消息
        if(open){
            ListenableFuture<SendResult<String, Object>> future = kafkaTemplate.send(topic, jsonObj);
            future.addCallback(new ListenableFutureCallback<SendResult<String, Object>>() {
                @Override
                public void onFailure(Throwable throwable) {
                    logger.info("Produce failed to be sent:" + throwable.getMessage());
                }
                @Override
                public void onSuccess(SendResult<String, Object> stringObjectSendResult) {
                    logger.info("Produce success result: " + stringObjectSendResult.toString());
                }
            });
        }else{
            logger.info("kafka 开关未开启！");
        }
    }
}
